package com.aotain.project.apollo.boltv2;

import java.net.URLDecoder;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.storm.AbstractBolt;

/**
 * 跨站脚本攻击检测
 * @author Administrator
 *
 */
public class XSSDetectBolt  extends AbstractBolt{

	/**
	 * 
	 */
	private static final long serialVersionUID = 8820683203058905439L;

	private Map<String,Integer> xssMap = new HashMap<String,Integer>();
	
	/**
	 * 上行包字节数
	 */
	private HashMap<String,Long> mapStreamoctets = new HashMap<String,Long>();
	
	/**
	 * 上行包数
	 */
	private HashMap<String,Long> mapStreampacket = new HashMap<String,Long>();
	
	private final int THRESHOLD = 1;
	
	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			if(isTickTuple(tuple)) {
				emitCountingData(collector);
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			collector.fail(tuple);
			throw new FailedException("XSSDetectBolt throws an exception!",e);
		}
	}
	
	//计算需要的值
	private void countInLocal(Tuple tuple) {
		try
		{
			String url = tuple.getStringByField("url").trim();	
			String refer = tuple.getStringByField("refer").trim();	
			if(!refer.trim().isEmpty()) {
				return;//XSS 是无法模拟出refer的
			}
			
			url = url.replace("%", "%25");
			if(matchRuleBase(url) == 2){
				String type =  tuple.getStringByField("type").trim();	
				String dip = tuple.getStringByField("dip");
				String dport = tuple.getStringByField("dport");
				String sip = tuple.getStringByField("sip");		
				String gis = new String(Base64.decodeBase64(tuple.getStringByField("gisstr")));
				String idc = tuple.getStringByField("idcno");
				
				String Key = dip + "|" + dport + "|" + sip + "|" + gis +  "|"  + type + "|"  + idc + "|" + url;
//				url = url.toLowerCase();				
				if(xssMap.containsKey(Key)) {				
					xssMap.put(Key, xssMap.get(Key) + 1);
				}else{
					xssMap.put(Key,1);
				}
				
				if(tuple.contains("upstreamoctets") && tuple.contains("upstreampacket"))
				{
					long nUpStreamOctets = tuple.getLongByField("upstreamoctets");
					long nUpstreampacket = tuple.getLongByField("upstreampacket");
					if(mapStreamoctets.containsKey(Key))
					{
						nUpStreamOctets = mapStreamoctets.get(Key) + nUpStreamOctets;
					}
					mapStreamoctets.put(Key, nUpStreamOctets);
					
					if(mapStreampacket.containsKey(Key))
					{
						nUpstreampacket = mapStreampacket.get(Key) + nUpstreampacket;
					}
					mapStreampacket.put(Key, nUpstreampacket);
				
				}
					//tupleMap.put(Key, dip + "|" + dport + "|" + sip + "|" + sport +"|" + gis +  "|"  + type + "|"  + idc);
			}
				
		}
		catch(Exception ex)
		{
			Logger.getRootLogger().error("=====countInLocal=========ERROR=================",ex);
		}
	}
	
	
	//定时发送
		private void emitCountingData(OutputCollector collector) {
			
			try{
				HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);
				String info = String.format("######DEBUG:XSS MAP[%d]",xssMap.size());
				Logger.getRootLogger().info(info);
				
				for(String dkey : xssMap.keySet()) {	
					int urlPVNum = xssMap.get(dkey);
					if(urlPVNum >= THRESHOLD){
						
						//String Key = dip + "|" + dport + "|" + sip + "|" + gis +  "|"  + type + "|"  + idc + "|" + url;
						String[] tuplesp = dkey.split("\\|",-1);
						String dip = tuplesp[0];
						String dport = tuplesp[1];
						String sip = tuplesp[2];
						String gis = tuplesp[3];
						String type = tuplesp[4];
						String idc = tuplesp[5];
						String url = tuplesp[6];
						String[] gisArray = gis.split("#");
						String destAreaName = gisArray[0];
						String destGis = gisArray[1];
						String sourceAreaName = gisArray[2];
						String sourceGis = gisArray[3];
						String sourceAreaCountry = gisArray[4];
						String sourceAreaId = gisArray[5];
						String sourceProvinceName =  gisArray[6].trim(); //如果省为空，精确到国家
						String sourceAreaCityId = gisArray[7];
						String sourceAreaProvinceId = gisArray[8];
					
							
						String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(System.currentTimeMillis()));
						String rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + dport + "_10";	
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEIP", sip);
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTPORT", dport);							
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ACCESSTIME", accesstime);
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ENDTIME", accesstime);
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ABRNORMAL", "10");
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTIP", dip);							
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEAREA", sourceAreaName);
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEGEO", sourceGis);
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCECOUNTRY", sourceAreaCountry);
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTAREA", destAreaName);
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTGEO", destGis);							
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ATTNUM", String.valueOf(urlPVNum));
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESC", "正在遭受  XSS跨站脚本攻击  " + url.replace("script", "#script#"));
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "EVALUATE", "40");	
						if(sourceProvinceName.equals("")) sourceProvinceName = sourceAreaCountry;
							hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "PROVINCE", sourceProvinceName);
							
						String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
					    String rowkey3 = String.format("%s_%s", accesstime, sip);
					    hbaseInstance.Add(tbName3, rowkey3, "cf", "ACCESSTIME", accesstime);
					    hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCEIP", sip);
					    hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCEGEO", String.format("%s",sourceGis));
					    hbaseInstance.Add(tbName3, rowkey3, "cf", "ATTNUM", String.valueOf(urlPVNum));
					    hbaseInstance.Add(tbName3, rowkey3, "cf", "AREAID", sourceAreaProvinceId);
					    hbaseInstance.Add(tbName3, rowkey3, "cf", "AREANAME", sourceProvinceName);
					    hbaseInstance.Add(tbName3, rowkey3, "cf", "CITYID", sourceAreaCityId);
					    hbaseInstance.Add(tbName3, rowkey3, "cf", "CITYNAME", sourceAreaName);
					    hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTIP", dip);
					    hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTPORT", dport);
					    hbaseInstance.Add(tbName3, rowkey3, "cf", "TYPE", "10");
					    hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCECOUNTRY", sourceAreaCountry);
					    hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTAREA", destAreaName);
					    hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTGEO", destGis);
					    
					  //UDII数据特有
						long nUpStreamOctets = 0;
						long nUpstreampacket = 0;
						if(mapStreamoctets.size() > 0 && mapStreampacket.size() > 0)
						{
							nUpStreamOctets = mapStreamoctets.get(dkey);
							nUpstreampacket = mapStreampacket.get(dkey);
							hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(nUpStreamOctets));
							hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(nUpstreampacket));
						}
					    
						Date current = new Date();
						SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
						SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
						SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
					    String dateStr_h = sdf_h.format(current);
						String dateStr_d = sdf_d.format(current);
						String dateStr_hour = sdf_hour.format(current);
						
						collector.emit(new Values(dip,sip,dport,"",dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,
					    		sourceProvinceName, 1l, dateStr_hour, "XSS" , idc, sourceAreaCountry,(long)nUpStreamOctets,(long)nUpstreampacket));
					    
//						collector.emit("STAT",new Values(dip,sip,"", "",
//				    			dateStr_d,dateStr_h,sourceProvinceID,sourceAreaName,
//				    			sourceProvinceName, nUpStreamOctets, dateStr_hour,"XSS", idc, sourceAreaCountry,(long)nUpStreamOctets,(long)nUpstreampacket));
//				   	
//				    	collector.emit("DELAY_",new Values(dip,sip,sourceAreaName, urlPVNum,
//				    			sourceAreaCountry, (long)nUpStreamOctets, (long)nUpstreampacket, "0"));
						
					}
				}
			}catch(Exception e){
				throw new FailedException("XSSDetectBolt throws an exception!",e);
			}finally{
				xssMap.clear();
			}
		}
	
		/**
		 * 匹配URL中存在 <script> </script> 字符串的数据
		 * @param url
		 * @return
		 */
		private static int  matchRuleBase(String url){
			
			int matchNum = 0;
			try{				
				if(url.isEmpty()) return matchNum;
				url = URLDecoder.decode(url,"utf-8").toLowerCase();
				String rgx = "<script>|</script>";
				Pattern pattern = Pattern.compile(rgx); 
				Matcher m = pattern.matcher(url);
				ArrayList<String> list = new ArrayList<String>(); 
				while(m.find()) {
					if(!list.contains(m.group())) 
						list.add(m.group());
				}
				matchNum =  list.size();
			}catch(Exception ex){
				Logger.getRootLogger().error("=====XSSDetectBolt=========ERROR=================" + url,ex);
			}		
			return matchNum;
		
		}

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare(new Fields("dip","sip","dport","sport","dateStr_d","dateStr_h","sourceAreaId","sourceAreaName","areaName","pv","dateStr_hour", "abntype", "idcno", "sourcecountry","streamoctets","streampacket"));
		declarer.declareStream("ABNORMAL_LOG", new Fields("rowkey","dip","sip","dport","accesstime","sourcearea",
				"sourcegeo","sourcecountry","destarea","destgeo","evaluate","abnormal","desc","attnum","delaytime","flag"));
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,60); //60秒处理一次
		return conf;
	}

	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		
	}
}
